package com.alibaba.nacos.common.remote.client.grpc;

import com.alibaba.nacos.api.grpc.auto.BiRequestStreamGrpc;
import com.alibaba.nacos.api.grpc.auto.Payload;
import com.alibaba.nacos.api.grpc.auto.RequestGrpc;
import com.alibaba.nacos.api.remote.request.ConnectionSetupRequest;
import com.alibaba.nacos.api.remote.request.Request;
import com.alibaba.nacos.api.remote.request.ServerCheckRequest;
import com.alibaba.nacos.api.remote.request.SetupAckRequest;
import com.alibaba.nacos.api.remote.response.Response;
import com.alibaba.nacos.api.remote.response.ServerCheckResponse;
import com.alibaba.nacos.api.remote.response.SetupAckResponse;
import com.alibaba.nacos.common.remote.client.Connection;
import com.alibaba.nacos.common.remote.client.RpcClient;
import com.alibaba.nacos.common.remote.client.ServerRequestHandler;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import io.grpc.CompressorRegistry;
import io.grpc.DecompressorRegistry;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.netty.shaded.io.grpc.netty.NegotiationType;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

@Slf4j
public abstract class GrpcClient
        extends RpcClient {

    private Executor executor = MoreExecutors.directExecutor();
    @Autowired
    private SetupRequestHandler setupRequestHandler;
    private StreamObserver<Payload> emitter;

    @Override
    public Connection connectToServer(ServerInfo serverInfo) {
        try {
            int port = serverInfo.getServerPort() + rpcPortOffset();
            ManagedChannel managedChannel = buildChannel(serverInfo.getServerIp(), port, null).executor(executor)
                    .compressorRegistry(CompressorRegistry.getDefaultInstance())
                    .decompressorRegistry(DecompressorRegistry.getDefaultInstance())
                    .maxInboundMessageSize(10 * 1024 * 1024)
                    .keepAliveTime(6 * 60 * 1000, TimeUnit.MILLISECONDS)
                    .keepAliveTimeout(20, TimeUnit.SECONDS)
                    .build();
            RequestGrpc.RequestFutureStub requestFutureStub = RequestGrpc.newFutureStub(managedChannel);

            ServerCheckRequest serverCheckRequest = new ServerCheckRequest();
            ListenableFuture<Payload> responseFuture = requestFutureStub.request(GrpcUtils.convert(serverCheckRequest));
            Payload responsePayload = responseFuture.get(3000L, TimeUnit.MILLISECONDS);
            Response Response = GrpcUtils.parse(responsePayload);
            if (!(Response instanceof ServerCheckResponse)) {
                managedChannel.shutdownNow();
                return null;
            }

            ServerCheckResponse serverCheckResponse = (ServerCheckResponse) Response;
            final String connectionId = serverCheckResponse.getConnectionId();

            BiRequestStreamGrpc.BiRequestStreamStub biRequestStreamStub = BiRequestStreamGrpc.newStub(managedChannel);
            emitter = biRequestStreamStub.requestBiStream(new StreamObserver<Payload>() {
                @Override
                public void onNext(Payload payload) {
                    Request request = GrpcUtils.parse(payload);
                    if (request == null) {
                        return;
                    }

                    if (request instanceof SetupAckRequest setupAckRequest) {
                        setupRequestHandler.requestReply(setupAckRequest, null);
                        return;
                    }

                    Response response = handleServerRequest(request);
                    if (response == null) {
                        log.warn("failed to process server request connectionId:{} ackId:{}", connectionId, request.getRequestId());
                    } else {
                        response.setRequestId(request.getRequestId());
                        send(response);
                    }
                }

                @Override
                public void onError(Throwable t) {
                    throw new UnsupportedOperationException(t);
                }

                @Override
                public void onCompleted() {
                    throw new UnsupportedOperationException();
                }
            });

            GrpcConnection grpcConn = new GrpcConnection().setRequestFutureStub(requestFutureStub).setChannel(managedChannel).setExecutor(executor);
            grpcConn.setConnectionId(connectionId);
            grpcConn.setServerInfo(serverInfo);
            if (serverCheckResponse.isSupportAbilityNegotiation()) {
                grpcConn.setAbilityTable(null);
            }

            ConnectionSetupRequest connectionSetupRequest = new ConnectionSetupRequest().setClientVersion("Nacos-Java-Client:v2.5.1")
                    .setLabels(Map.of("source", "sdk", "module", "naming"))
                    .setAbilityTable(null)
                    .setTenant(null);
            send(connectionSetupRequest);
            log.info("connectToServer id:{} server address: {}", connectionId, serverInfo.getServerIp() + ":" + port);
            return grpcConn;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }

    private void send(Response response) {
        this.emitter.onNext(GrpcUtils.convert(response));
    }

    private void send(Request request) {
        this.emitter.onNext(GrpcUtils.convert(request));
    }

    private ManagedChannelBuilder buildChannel(String serverIp,
                                               int port,
                                               SslContext sslContext) {
        NettyChannelBuilder builder = NettyChannelBuilder.forAddress(serverIp, port);
        if (sslContext != null) {
            builder.negotiationType(NegotiationType.TLS).sslContext(sslContext);
        } else {
            builder.usePlaintext();
        }
        return builder;
    }

    @Component
    public static class SetupRequestHandler
            implements ServerRequestHandler {

        @Override
        public Response requestReply(Request request,
                                     Connection connection) {
            if (request instanceof SetupAckRequest setupAckRequest) {
                return new SetupAckResponse();
            }
            return null;
        }
    }

    static class RecAbilityContext {

    }
}
